package com.atguigu.gmall.realtime.app.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {

//        生命周期方法用于初始化连接  Phoenix 创建
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
////        设置为自动提交  如果批量提交就不用自动 手动Ccommit
//        connection.setAutoCommit(true);
    }

    //    Value 格式
//      SQL 格式
    @Override
    public void invoke(JSONObject value, Context context) throws SQLException {
        PreparedStatement preparedStatement = null;
        try {
//          获取sql语句
            String sinkTable = value.getString("sinkTable");

            JSONObject after = value.getJSONObject("after");
            String upsertSql = getUpserSql(sinkTable,
                    after
            );
//        预编译sql
            preparedStatement = connection.prepareStatement(upsertSql);

//            判断如果当前数据为更新操作，则先删除Redis中的数据 OR 更新Redis的数据 防止删除过程中被读取  维度数据变化缓慢
            if("update".equals(value.get("type"))) {
//                Redis 删除一个不存在的Key是不会报错的
                DimUtil.delRedisDimInfo(sinkTable.toUpperCase(), after.getString("id"));
            }

//        执行插入操作
            preparedStatement.executeUpdate();
            connection.commit();
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }

        }
    }


    private String getUpserSql(String sinkTable, JSONObject data) {

        Set<String> ketSet = data.keySet();
        Collection<Object> values = data.values();

//        keySet.MKsTRING(",") => "id,tm_name"
//        Hbase 底层都是varchar

        return "upsert into " + GmallConfig.HBASE_SCHEMA + "." + sinkTable + "(" +
//                value ('','','','')
                StringUtils.join(ketSet, ",") + ") values ('" +
                StringUtils.join(values, "','") + "')";
    }
}
